use dashmap::DashMap;
use dashmap::mapref::one::Ref;
use rumqttc::{EventLoop, Publish};
use std::sync::Arc;

#[async_trait::async_trait]
pub trait PacketHandler: Send + Sync {
    async fn handle(&self, packet: Publish);
}

pub struct PacketHandlerStore {
    inner: Arc<DashMap<String, Arc<dyn PacketHandler>>>,
}

impl<'a> PacketHandlerStore {
    pub fn new() -> Self {
        PacketHandlerStore {
            inner: Arc::new(DashMap::new()),
        }
    }

    pub fn store(&self, topic: String, handler: Arc<dyn PacketHandler>) {
        self.inner.insert(topic, handler);
    }

    pub fn load(&'a self, topic: &str) -> Option<Ref<'a, String, Arc<dyn PacketHandler>>> {
        self.inner.get(topic)
    }

    pub fn delete(&self, topic: String) {
        self.inner.remove(&topic);
    }
}

pub struct EventLoopTask {
    event_loop: EventLoop,
    handler_store: Arc<PacketHandlerStore>,
}

impl EventLoopTask {
    pub fn new(event_loop: EventLoop, handler_store: Arc<PacketHandlerStore>) -> Self {
        Self {
            event_loop,
            handler_store,
        }
    }

    pub async fn run(&mut self) {
        loop {
            match self.event_loop.poll().await {
                Ok(event) => match event {
                    rumqttc::Event::Incoming(rumqttc::Packet::Publish(publish_packet)) => {
                        log::debug!(
                            "接收到消息，topic '{}': {}",
                            publish_packet.topic,
                            String::from_utf8_lossy(&publish_packet.payload)
                        );
                        if let Some(handler) = self.handler_store.load(&publish_packet.topic) {
                            let handler = handler.clone();
                            tokio::spawn(async move {
                                handler.handle(publish_packet).await;
                            });
                        }
                    }
                    _ => {
                        log::debug!("其他事件: {:?}", event);
                    }
                },
                Err(err) => {
                    log::error!("{:?}", err);
                }
            }
        }
    }
}
